package my.algo.data;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author fengbo
 * @date 2018/4/16
 */
public class BlockingQueue<T> {

    private final Object[] queue;

    private final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;

    private int count;
    private int putIndex;
    private int takeIndex;

    private BlockingQueue(int initialCapacity) {
        lock = new ReentrantLock(false);
        notEmpty = lock.newCondition();
        notFull = lock.newCondition();
        queue = new Object[initialCapacity - 1];
    }

    private boolean put(T element) throws InterruptedException {
        ReentrantLock lock = this.lock;
        lock.lock();
        try {
            while (count == queue.length) {
                notFull.await();
            }
            queue[putIndex] = element;
            if (++putIndex == queue.length) {
                putIndex = 0;
            }
            count++;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
        return true;
    }

    private T take() throws InterruptedException {
        ReentrantLock lock = this.lock;
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            @SuppressWarnings("unchecked")
            T result = (T) queue[takeIndex];
            queue[takeIndex] = null;
            if (++takeIndex == queue.length) {
                takeIndex = 0;
            }
            count--;
            notFull.signal();
            return result;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new BlockingQueue<>(4);
        new Thread(() -> {
            try {
                for (int i = 0; i < 100; i++) {
                    System.out.println(Thread.currentThread().getName() + "aa" + i);
                    queue.put("aa" + i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        Thread.sleep(100);

        new Thread(() -> {
            try {
                System.out.println(queue.take());
                System.out.println(queue.take());
                System.out.println(queue.take());
                System.out.println(queue.take());
                System.out.println(queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
        Thread.sleep(100);
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
    }
}
